package com;

import com.handler.client.RpcRespHandler;
import com.message.RpcReqMessage;
import com.message.factory.RpcMessageFactory;
import com.protocol.SharedMessageCodec;
import com.service.HelloService;
import com.service.RpcPromiseContentFactory;
import com.utils.SequenceIdGenerator;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

/**
 * @author 罗俊华
 * @date 2021/9/1 - 5:01 下午
 */
@Slf4j
public class Client {

    private static volatile Channel channel;

    private static final RpcPromiseContentFactory RPC_PROMISE_CONTENT_FACTORY = new RpcPromiseContentFactory();

    public static void main(String[] args) throws InterruptedException {


        HelloService helloService = getTargetProxy(HelloService.class);

        String luoHello = helloService.sayHello("罗俊华");

        String gkyHello = helloService.sayHello("gky");

        System.out.println(luoHello);

        System.out.println(gkyHello);
    }

    @SuppressWarnings("代理安全")
    public static  <T> T getTargetProxy(Class<T> interfaceClass) {

        // 代理类的类加载器和接口的类加载器可以这样设置为同一个
        Object result = Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class[]{interfaceClass}, new InvocationHandler() {

            /**
             * 将来代理类的任何一个方法的执行都会进入到 invoke() 方法中执行
             *
             * @param proxy 代理对象
             * @param method 当前代理类正在执行的方法
             * @param args 当前正在执行的方法的实际参数
             * @return
             * @throws Throwable
             */
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {


                int sequenceId = SequenceIdGenerator.getSequenceId();
                RpcReqMessage rpcReqMessage = new RpcReqMessage(
                        sequenceId,
                        interfaceClass,
                        method.getName(),
                        method.getParameterTypes(),
                        args
                );

//                获取目标接口的返回类型
                Class<?> returnType = method.getReturnType();

//                发送 Rpc 调用请求
                getChannel().writeAndFlush(rpcReqMessage);

//                创建一个 promise 对应要用来通过线程异步接收 结果
                DefaultPromise<Object> promise = new DefaultPromise<>(getChannel().eventLoop());

//                将接收rpc结果的容器放入 map中
                RPC_PROMISE_CONTENT_FACTORY.getPromiseMap().put(sequenceId,promise);

//                等待 rpc 响应
                Promise<Object> resultPromise = promise.await();

                if(resultPromise.isSuccess()){

//                将响应转换为方法的返回类型
                    return returnType.cast(resultPromise.get());
                }

                throw new RuntimeException(resultPromise.cause());
            }
        });

//        将生成的代理类转换为 被代理的目标接口类型
        return interfaceClass.cast(result); //(T) result;

    }


    /**
     * 双检锁与 volatile 实现单例
     *
     * @return
     * @throws InterruptedException
     */
    public static Channel getChannel() throws InterruptedException {

        if (channel != null) {
            return channel;
        }
        synchronized (Client.class) {

            if (channel != null) {
                return channel;
            }
            initChannel();
        }
        return channel;
    }

    private static void initChannel() {
        NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup();

        SharedMessageCodec SHARED_MESSAGE_CODEC = new SharedMessageCodec(new RpcMessageFactory());

        LoggingHandler LOGGING_HANDLER = new LoggingHandler();

        /**
         * 该 handler 因为里面有 Map，那么必须是共享的
         * 不然rpc调用线程和rpc响应线程将不能通信
         */
        RpcRespHandler RPC_RESP_HANDLER = new RpcRespHandler(RPC_PROMISE_CONTENT_FACTORY.getPromiseMap());


        try {
            Client.channel = new Bootstrap()
                    .group(nioEventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel ch) throws Exception {
                            ch.pipeline().addLast(
                                    new LengthFieldBasedFrameDecoder(1024, 11, 4),
                                    SHARED_MESSAGE_CODEC,
                                    LOGGING_HANDLER,
                                    RPC_RESP_HANDLER
                            );
                        }
                    })
                    .connect("localhost", 10111)
                    .sync()
                    .channel();
        } catch (InterruptedException e) {
            e.printStackTrace();
            nioEventLoopGroup.shutdownGracefully();
        }
    }
}
